package cn.xval.ft.core.marketdata;

import cn.xval.ft.core.marketdata.exceptions.MdApiException;
import cn.xval.ft.core.marketdata.manage.DataWatcher;
import cn.xval.ft.core.marketdata.manage.SpiManager;
import cn.xval.ft.core.marketdata.manage.MarketDataSpi;
import cn.xval.ft.grpc.*;
import com.google.protobuf.Empty;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class MarketDataService extends MarketDataGrpc.MarketDataImplBase {
    @Override
    public void start(StartCollectRequest request, StreamObserver<Empty> responseObserver) {
        for (var info : request.getCollectionsList() ) {
            var spi = new MarketDataSpi(info);
            try {
                log.info("建立链接: {}:{}@{}", info.getUserId(), info.getBrokerId(), info.getHost());
                spi.start();
            } catch (MdApiException e) {
                log.error("建立链接失败", e);
            }
            SpiManager.addSpi(spi);
        }
        responseObserver.onCompleted();
    }

    @Override
    public void subscribe(MarketDataAddWatchRequest request, StreamObserver<Empty> responseObserver) {
        SpiManager.subscribeMarketData(request.getConnectionId(), request.getContactCodesList());
        responseObserver.onCompleted();
    }

    @Override
    public void watch(Empty request, StreamObserver<MarketDataInfo> responseObserver) {
        new DataWatcher(responseObserver);
    }

    @Override
    public void getStatus(Empty request, StreamObserver<CollectorStatus> responseObserver) {
        responseObserver.onCompleted();
    }

    @Override
    public void stop(StopCollectRequest request, StreamObserver<Empty> responseObserver) {
        responseObserver.onCompleted();
    }
}
